package com.yjxxt;


import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class Hello01Consumer extends Thread {

    //创建消费者对象
    private KafkaConsumer<String, String> consumer;

    /**
     * 创建构造器
     */
    public Hello01Consumer(String cname) {
        super.setName(cname);
        //读取配置文件
        Properties properties = new Properties();
        //ZK地址
        properties.put("bootstrap.servers", "192.168.255.11:9092,192.168.255.12:9092,192.168.255.13:9092");
        //消费者所在组的名称
        properties.put("group.id", "yjx-bigdata");
        //ZK超时时间
        properties.put("zookeeper.session.timeout.ms", "1000");
        //反序列化
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //当消费者第一次消费时，从最低的偏移量开始消费
        properties.put("auto.offset.reset", "earliest");
        //自动提交偏移量
        properties.put("auto.commit.enable", "true");
        //消费者自动提交偏移量的时间间隔
        properties.put("auto.commit.interval.ms", "1000");
        //创建消费者对象
        consumer = new KafkaConsumer<>(properties);
    }

    @Override
    public void run() {
        try {
            consumer.subscribe(Arrays.asList("userlog"));
            boolean flag = true;
            while (flag) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (TopicPartition partition : records.partitions()) {
                    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                    for (ConsumerRecord<String, String> record : partitionRecords) {
                        System.out.println(record.topic() + ":" + record.partition() + ":" + record.offset() + ":" + record.key() + ":" + record.value());
                    }
                    //同步
                    consumer.commitSync();
                }
            }
        } finally {
            consumer.close();
        }
    }

    public static void main(String[] args) {
        Hello01Consumer consumer01 = new Hello01Consumer("马龙YYDS");
        consumer01.start();
    }
}